package com.spx.chapter07;

import com.spx.chapter05.pojo.Event;
import com.spx.chapter06.LateDataTest;
import com.spx.chapter06.entity.UrlViewCount;
import com.spx.util.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;

/**
 * create by undeRdoG on  2022-05-03  15:52
 * 凡心所向，素履以往，生如逆旅，一苇以航。
 */
public class TopNExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );

        // 按照url分组，统计窗口内每个url的访问量
        SingleOutputStreamOperator<UrlViewCount> urlCountStream = stream.keyBy(data -> data.url)
                // 开一个滚动窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .aggregate(new LateDataTest.CustomAggregateFunction(), new LateDataTest.CustomProcessWindowFunction());


        urlCountStream.print("urlCountStream");

        urlCountStream.keyBy(data -> data.end)
                .process(new TopNProcessResult(2)).setMaxParallelism(6)
                .print("res");


        // 对于同一窗口统计的访问量，进行收集排序


        env.execute();
    }


    /**
    * 实现自定义的KeyedProcessFunction
    * */
    public static class TopNProcessResult extends KeyedProcessFunction<Long,UrlViewCount,String>{

        private Integer topNum;

        // 定义列表状态
        private ListState<UrlViewCount> UrlViewCountState;

        public TopNProcessResult(Integer topNum) {
            this.topNum = topNum;
        }

        // 在运行环境中获取状态，相当于把该状态托管到Flink中
        @Override
        public void open(Configuration parameters) throws Exception {
            this.UrlViewCountState = getRuntimeContext().getListState(
                    new ListStateDescriptor<UrlViewCount>("url-count-list", Types.POJO(UrlViewCount.class)));
        }

        @Override
        public void processElement(UrlViewCount value, Context ctx, Collector<String> out) throws Exception {
            // 将数据保存到状态中
            UrlViewCountState.add(value);

            // 注册定时器 即 windowEnd + 1ms 的定时器
            ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey() + 1);
        }


        // 定时器的回调
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

            List<UrlViewCount> urlViewCountArrayList = new ArrayList<>();

            Iterable<UrlViewCount> urlViewCounts = this.UrlViewCountState.get();
            Iterator<UrlViewCount> iterator = urlViewCounts.iterator();
            while (iterator.hasNext()){
                urlViewCountArrayList.add(iterator.next());
            }

            urlViewCountArrayList.sort(new Comparator<UrlViewCount>() {
                @Override
                public int compare(UrlViewCount o1, UrlViewCount o2) {
                    // 升序排序
                   if (o1.count.equals(o2.count)){
                       return 0;
                   }else if (o1.count > o2.count){
                       return -1;
                   }else {
                       return 1;
                   }
                }
            });

            StringBuilder builder = new StringBuilder();
            builder.append("--------------------------------------------").append("\n");
            builder.append("window finish time is " + urlViewCountArrayList.get(0).end).append("\n");
            for (int i = 0; i < this.topNum; i++){
                UrlViewCount urlViewCount = urlViewCountArrayList.get(i);
                builder.append(String.format("url is %s and count is %s", urlViewCount.url,urlViewCount.count)).append("\n");

            }

            out.collect(builder.toString());
        }
    }

}
